package tsif.tcluster.netty.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.io.DataInput;
import java.io.IOException;
import java.util.List;

import tsif.tcluster.netty.impl.NettyCodecUtils;
import tsif.tcluster.netty.impl.NettyCodecUtils.IDecode;

import com.esotericsoftware.minlog.Log;

/**
 * 协议解码器<br>
 * 
 */
public class TClusterNettyDecoder extends ByteToMessageDecoder {

	/** 解析器 **/
	private static IDecode<TClusterNettyMessage> decode = new IDecode<TClusterNettyMessage>() {

		@Override
		public TClusterNettyMessage read(DataInput inBuf) throws IOException {
			TClusterNettyMessage msg = new TClusterNettyMessage();
			msg.setSourceServiceId(inBuf.readLong());
			msg.setTargetServiceId(inBuf.readLong());
			msg.setCode(inBuf.readShort());

			// read data
			int datalen = inBuf.readShort(); // 读取数据长度
			byte[] data = new byte[datalen];
			inBuf.readFully(data);
			msg.setData(data);
			return msg;
		}

	};

	@Override
	protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		// 循环解码
		while (true) {
			try {
				Object obj = decode(ctx, in);
				if (obj == null) {
					break; // 解析不出來
				}
				out.add(obj);
			} catch (Exception e) {
				Log.error("解析消息错误! from=" + ctx.channel().remoteAddress(), e);
				close(ctx);
			}
		}
	}

	/** 解码处理 **/
	private Object decode(ChannelHandlerContext ctx, ByteBuf in) throws IOException {
		return NettyCodecUtils.decode(in, decode);
	}

	/** 关闭连接 **/
	private void close(ChannelHandlerContext ctx) {
		ctx.channel().close();
	}

}
